package us.codecraft.webmagic.core;

import us.codecraft.webmagic.core.downloader.HttpClientDownloader;
import us.codecraft.webmagic.core.pipeline.ConsolePipeline;
import us.codecraft.webmagic.core.processor.PageProcessor;
import us.codecraft.webmagic.core.thread.CountableThreadPool;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

/**
 * @author <p><a>fishlikewater@126.com</a></p>
 * @date 2019年08月11日 13:37
 * @since
 **/
public class XSpider extends Spider {

    /** 2019/01/23 by zhangx */
    private ComplateListener complateListener;

    private Consumer<XSpider> errorHand;

    /**
     * create a spider with pageProcessor.
     *
     * @param pageProcessor pageProcessor
     */
    public XSpider(PageProcessor pageProcessor) {
        super(pageProcessor);
    }

    public static XSpider create(PageProcessor pageProcessor) {
        return new XSpider(pageProcessor);
    }

    protected void initComponent() {
        if (downloader == null) {
            this.downloader = new HttpClientDownloader();
        }
        if (pipelines.isEmpty()) {
            pipelines.add(new ConsolePipeline());
        }
        downloader.setThread(threadNum);
        if(!downloader.isOpenAsync()){
            if (threadPool == null || threadPool.isShutdown()) {
                if (executorService != null && !executorService.isShutdown()) {
                    threadPool = new CountableThreadPool(threadNum, executorService);
                } else {
                    threadPool = new CountableThreadPool(threadNum);
                }
            }
        }
        if (startRequests != null) {
            for (Request request : startRequests) {
                addRequest(request);
            }
            startRequests.clear();
        }
        startTime = new Date();
    }

    @Override
    public void run(){
        try {
            checkRunningStat();
            initComponent();
            logger.info("Spider {} started!",getUUID());
            if(downloader.isOpenAsync()){
                startDowanloadAsync();
            }else {
                startDowanloadSync();
            }
            stat.set(STAT_STOPPED);
            // release some resources
            if (destroyWhenExit) {
                close();
            }
            excutorComplateListener(this);
            logger.info("Spider {} closed! {} pages downloaded.", getUUID(), getPageCount());
        }catch (Exception e){
            excutorErrorHand(e);
        }finally {

        }
    }


    private void startDowanloadAsync(){
        while (!Thread.currentThread().isInterrupted() && stat.get() == STAT_RUNNING) {
            final Request request = scheduler.poll(this);
            if (request == null) {
                if (downloader.isComplete() && exitWhenComplete) {
                    break;
                }
                // wait until new url added
                waitAsyncNewUrl();
            } else {
                downloader.asyncDowanload(request, this);
            }
        }
    }

    private void startDowanloadSync(){
        while (!Thread.currentThread().isInterrupted() && stat.get() == STAT_RUNNING) {
            final Request request = scheduler.poll(this);
            if (request == null) {
                if (threadPool.getThreadAlive() == 0 && exitWhenComplete) {
                    break;
                }
                // wait until new url added
                waitNewUrl();
            } else {
                threadPool.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            processRequest(request);
                            onSuccess(request);
                        } catch (Exception e) {
                            onError(request);
                            logger.error("process request " + request + " error", e);
                        } finally {
                            pageCount.incrementAndGet();
                            signalNewUrl();
                        }
                    }
                });
            }
        }
    }

    protected void waitAsyncNewUrl() {
        newUrlLock.lock();
        try {
            //double check
            if (downloader.isComplete() && exitWhenComplete) {
                return;
            }
            newUrlCondition.await(emptySleepTime, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            logger.warn("waitAsyncNewUrl - interrupted, error {}", e);
        } finally {
            newUrlLock.unlock();
        }
    }

    /**
     *  It is turned off by default and can be set to keep connection pool after each grab in some frequent operations.<br><br>
     * @param destroyWhenExit Do you want to close connection pool
     * @return Spider
     * 2019/01/23 by zhangx
     */
    public XSpider setDestroyWhenExit(boolean destroyWhenExit){
        this.destroyWhenExit = destroyWhenExit;
        return this;
    }


    /**
     * @see ComplateListener
     * @param complateListener defined listener
     * @return Spider
     * 2019/01/23 by zhangx
     */
    public XSpider setComplateListener(ComplateListener<Spider> complateListener){
        this.complateListener = complateListener;
        return this;
    }

    public XSpider setErrorHand(Consumer<XSpider> consumer){
        this.errorHand = consumer;
        return this;
    }


    /**
     * When asynchronous fetching finishes executing the listener.<br><br>
     * 2019/01/23 by zhangx
     */
    private void excutorComplateListener(Spider spider){
        if(complateListener != null){
            complateListener.excutor(spider);
        }
    }

    private void excutorErrorHand(Exception e) {
        if(errorHand != null){
            errorHand.accept(this);
        }else {
            logger.error("Spider process has a error", e);
        }
    }

    public AtomicLong pageCount(){
        return pageCount;
    }
}
